package master

import (
	"fmt"
	"mapreduce/common"
	"mapreduce/worker"
	"sync"
)

var node common.Node

func init() {
	node.Addr = "localhost:8011"
	node.Id = 1
}

/**
  客户端提交Job
*/
func SubmitJob(job common.Job) {
	//fmt.Printf("开始执行 %s 任务，Map数量：%d, Reduce数量：%d\n", job.JobName, strconv.Itoa(job.MapNum), strconv.Itoa(job.ReduceNum))
	schedule(job)
}

/*
  Worker的Map和Reduce调度
*/
func schedule(job common.Job) {
	var mapWG sync.WaitGroup
	reduceNumberKeyFilePathSlcMap := make(map[int][]string)
	var lock sync.Mutex
	fmt.Printf("Map数量:%v\n", job.MapNum)
	for i := 0; i < job.MapNum; i++ {
		//必须放在go外面，不然不等待
		mapWG.Add(1)
		go func(mapJob common.Job, mapNumber int) {
			mapJob.TaskNumber = mapNumber
			fmt.Printf("Master：Map编号:%v\n", mapJob.TaskNumber)
			reduceNumberKeyFilePathMap := worker.ReceiveTask(mapJob, common.Map_Phase)
			fmt.Printf("Map编号：%v,执行完毕返回的文件路径：%v\n", mapNumber, reduceNumberKeyFilePathMap)
			for reduceNumber, filePath := range reduceNumberKeyFilePathMap {
				if _, ok := reduceNumberKeyFilePathMap[reduceNumber]; ok {
					reduceFilePathSlc := reduceNumberKeyFilePathSlcMap[reduceNumber]
					reduceFilePathSlc = append(reduceFilePathSlc, filePath)
					//这行需要加，不然使用的还是原来的reduceFilePathSlc--被释放了
					lock.Lock()
					reduceNumberKeyFilePathSlcMap[reduceNumber] = reduceFilePathSlc
					lock.Unlock()
				} else {
					reduceFilePathSlc := make([]string, 1)
					reduceFilePathSlc[0] = filePath
					lock.Lock()
					reduceNumberKeyFilePathSlcMap[reduceNumber] = reduceFilePathSlc
					lock.Unlock()
				}
			}
			mapWG.Done()
		}(job, i)
	}
	mapWG.Wait()
	for reduceNumber, reduceFilePathSlc := range reduceNumberKeyFilePathSlcMap {
		fmt.Printf("map生成的文件：reduce编号：%d, 生成的文件路径：%v\n", reduceNumber, reduceFilePathSlc)
	}
	resultFilePathMap := make(map[int]string)
	var reduceWG sync.WaitGroup
	fmt.Printf("开始执行Reduce,Map生成的文件切片：%v\n", reduceNumberKeyFilePathSlcMap)
	for i := 0; i < job.ReduceNum; i++ {
		reduceWG.Add(1)
		fmt.Printf("Reduce执行的编号：%v\n", i)
		go func(reduceNumber int) {
			//值不能直接使用i，需要传递进来
			job.TaskNumber = reduceNumber
			job.TaskFilePathSlc = reduceNumberKeyFilePathSlcMap[reduceNumber]
			reduceFilePathMap := worker.ReceiveTask(job, common.Reduce_Phase)
			lock.Lock()
			resultFilePathMap[reduceNumber] = reduceFilePathMap[reduceNumber]
			lock.Unlock()
			reduceWG.Done()
		}(i)
	}
	reduceWG.Wait()
	for reduceNumber, resultFilePath := range resultFilePathMap {
		fmt.Printf("输出的Reduce任务编号：%d,文件地址：%v\n", reduceNumber, resultFilePath)
	}
	fmt.Println("MapReduce执行完毕")
}
